package 实训二;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import com.bw.utils.DateFormatUtil;
import com.bw.utils.MyKafkaUtil;
import org.apache.flink.api.common.functions.FlatMapFunction;
import org.apache.flink.api.common.functions.RichFilterFunction;
import org.apache.flink.api.common.state.StateTtlConfig;
import org.apache.flink.api.common.state.ValueState;
import org.apache.flink.api.common.state.ValueStateDescriptor;
import org.apache.flink.api.common.time.Time;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.KeyedStream;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.connectors.kafka.FlinkKafkaConsumer;
import org.apache.flink.streaming.connectors.kafka.FlinkKafkaProducer;
import org.apache.flink.util.Collector;

/*
1.过滤页面数据中的独立访客访问记录
1.去重减少数据量
2.使用状态编程  过滤出独立访客
3，给状态设置生命周期24

1.加载数据流  页面
2.etl
3.减少数据流
4.保留当天第一天数据
5.清空状态
6.存入到kafka中
 */
public class DwdUv {
    public static void main(String[] args) throws Exception {
        // 环境
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);

        // 从kafka dwd_traffic_page_log 主题读取日志数据 封装为流
        String topic = "dwd_traffic_page_log";
        String groupId = "dwd_traffic_user_jump_detail";

        FlinkKafkaConsumer<String> kafkaConsumer = MyKafkaUtil.getFlinkKafkaConsumer(topic, groupId);
        DataStreamSource<String> pageLog = env.addSource(kafkaConsumer);

        pageLog.print();

        // 转换结构
        SingleOutputStreamOperator<JSONObject> mappedStream = pageLog.flatMap(new FlatMapFunction<String, JSONObject>() {
            @Override
            public void flatMap(String s, Collector<JSONObject> collector) throws Exception {
                try {
                    JSONObject jsonObject = JSON.parseObject(s);
                    collector.collect(jsonObject);
                } catch (Exception e) {
                    System.out.println("脏数据" + s);
                }
            }
        });

        // 过滤last_page_id 不为 null 的数据
        SingleOutputStreamOperator<JSONObject> firstPageStream = mappedStream.filter(
                jsonObject -> jsonObject
                        .getJSONObject("page")
                        .getString("last_page_id") == null
        );

        // 按照mid分组
        KeyedStream<JSONObject, String> keyedStream =
                firstPageStream.keyBy(jsonObject -> jsonObject.getJSONObject("common").getString("mid"));

        // 通过Flink状态编程过滤独立访客记录
        SingleOutputStreamOperator<JSONObject> filteredStream = keyedStream.filter(new RichFilterFunction<JSONObject>() {
            private ValueState<String> lastVisitDt;

            @Override
            public void open(Configuration parameters) throws Exception {
                super.open(parameters);
                ValueStateDescriptor<String> stateDescriptor = new ValueStateDescriptor<>("last_visit_dt", String.class);
                stateDescriptor.enableTimeToLive(
                        // 生命周期  如果某个用户超过一天没访问  改状态自动清除
                        StateTtlConfig.newBuilder(Time.days(1))
                                .setUpdateType(StateTtlConfig.UpdateType.OnCreateAndWrite)
                                .build()
                );
                this.lastVisitDt = getRuntimeContext().getState(stateDescriptor);
            }

            @Override
            public boolean filter(JSONObject jsonObject) throws Exception {
                // 获取当前访问日期
                String visitDt = DateFormatUtil.toDate(jsonObject.getLong("ts"));
                // 获取上次访问日期
                String lastDt = lastVisitDt.value();
                // 判断是否当天访问
                // 用户从未访问过保留此记录 更新状态
                if (lastDt != null || !lastDt.equals(visitDt)) {
                    lastVisitDt.update(visitDt);
                    return true;
                }
                // 今天已经访问过了就不再录入了
                return false;
            }
        });

        // 将独立访客数据写入到 kafka
        String targetTopic = "dwd_traffic_unique_visitor_detail";
        filteredStream.print(">>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>");

        FlinkKafkaProducer<String> kafkaProducer = MyKafkaUtil.getFlinkKafkaProducer(targetTopic);
        filteredStream.map(s->s.toJSONString()).addSink(kafkaProducer);

        // 提交作业到集群执行
        env.execute();

    }
}
